package com.heima.article.job;

import com.alibaba.fastjson.JSON;
import com.heima.article.service.HotArticleService;
import com.heima.common.constants.article.HotArticleConstants;
import com.heima.common.constants.mess.app.UpdateArticleMess;

import com.heima.model.mess.app.ArticleVisitStreamMess;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.BinaryOperator;
import java.util.stream.Collectors;

@Slf4j
@Component
public class ComputeHotArticleJob {
    @Resource
    HotArticleService hotArticleService;
    @Autowired
    StringRedisTemplate redisTemplate;


    @XxlJob("computeHotArticleJob")
    public ReturnT<String> computeHotArticleJob(String param){
        log.info("参数{}",param);
        log.info("热文章分值计算调度任务开始执行....");
        // 实现热点文章分值计算任务
        hotArticleService.computeHotArticle();
        log.info("热文章分值计算调度任务完成....");
        return ReturnT.SUCCESS;
    }

    @XxlJob("updateHotArticleJob")
    public ReturnT<String> updateHotArticleJob(String param){
        log.info("热文章分值更新 调度任务开始执行....");
        // 定时更新文章热度
//        统计redis队列数据
        Map<Long, List<ArticleVisitStreamMess>> listMap = listHandler();
        if (listMap == null){
            log.info("redis中无数据");
            return ReturnT.SUCCESS;
        }
//        将用户行为数据保存进数据库
        updateApArticle(listMap);
        log.info("热文章分值更新 调度任务完成....");
        return ReturnT.SUCCESS;
    }

    /**
     * 统计用户行为数据，存入数据库
     * @param listMap
     */
    private void updateApArticle(Map<Long, List<ArticleVisitStreamMess>> listMap) {
        listMap.forEach(((articleId, Messes) -> {
//            归并统计用户行为list
            Optional<ArticleVisitStreamMess> reduce = Messes.stream().reduce(new BinaryOperator<ArticleVisitStreamMess>() {
                @Override
                public ArticleVisitStreamMess apply(ArticleVisitStreamMess a1, ArticleVisitStreamMess a2) {
                    a1.setLike(a1.getLike() + a2.getLike());
                    a1.setComment(a1.getComment() + a2.getComment());
                    a1.setView(a1.getView() + a2.getView());
                    a1.setCollect(a1.getCollect() + a2.getCollect());
                    return a1;
                }
            });
            ArticleVisitStreamMess mess = reduce.get();
//            更新热点文章分值
            hotArticleService.updateApArticle(mess);
        }));
    }


    /**
     * 将各个微服务发送汇总到redis list中的数据进行处理
     */
    private Map<Long, List<ArticleVisitStreamMess>> listHandler() {
//        获取list的数据
        ListOperations<String, String> opsForList = redisTemplate.opsForList();
        Long size = opsForList.size(HotArticleConstants.HOT_ARTICLE_SCORE_BEHAVIOR_LIST);
        if (size == null || size.equals(0L)){
            return null;
        }
//        使用管道命令统一执行
        List<Object> objects = redisTemplate.executePipelined(new RedisCallback<List<UpdateArticleMess>>() {
            @Override
            public List<UpdateArticleMess> doInRedis(RedisConnection connection) throws DataAccessException {
//                从队列左侧获取size数量的元素
                connection.lRange(HotArticleConstants.HOT_ARTICLE_SCORE_BEHAVIOR_LIST.getBytes(), 0, size - 1);
//                截取队列剩下的元素
                connection.lTrim(HotArticleConstants.HOT_ARTICLE_SCORE_BEHAVIOR_LIST.getBytes(), size, -1);
                return null;
            }
//            设置序列化形式
        }, RedisSerializer.string());
        List<String> strings = (List<String>) objects.get(0);
//        解析管道命令返回结果
        List<UpdateArticleMess> messList = strings.stream()
                .map(o -> JSON.parseObject(o.toString(), UpdateArticleMess.class))
                .collect(Collectors.toList());
//        将 List<UpdateArticleMess> 映射为 List<ArticleVisitStreamMess> 并按ArticleId分组
        Map<Long, List<ArticleVisitStreamMess>> collect = messList.stream().map(updateArticleMess -> {
            ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
            articleVisitStreamMess.setArticleId(updateArticleMess.getArticleId());
            switch (updateArticleMess.getType()) {
                case LIKES:
                    articleVisitStreamMess.setLike(updateArticleMess.getAdd());
                    break;
                case VIEWS:
                    articleVisitStreamMess.setView(updateArticleMess.getAdd());
                    break;
                case COMMENT:
                    articleVisitStreamMess.setComment(updateArticleMess.getAdd());
                    break;
                case COLLECTION:
                    articleVisitStreamMess.setCollect(updateArticleMess.getAdd());
                    break;
            }
            return articleVisitStreamMess;
        }).collect(Collectors.groupingBy(ArticleVisitStreamMess::getArticleId));
        return collect;
    }

}
